package com.base.test.mysql2es.repository;

import com.base.test.mysql2es.model.ChildMapping;
import com.base.test.mysql2es.model.IncrementStorageType;
import com.base.test.mysql2es.model.Relation;
import com.base.test.mysql2es.util.*;
import com.google.common.base.Joiner;
import com.google.common.collect.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.common.UUIDs;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Component
@RequiredArgsConstructor
@SuppressWarnings("DuplicatedCode")
public class DataRepository {

    private static final String EQUALS_SUFFIX = "<-_->";
    private static final String EQUALS_I_SPLIT = "<=_=>";
    private static final Date NIL_DATE_TIME = new Date(0L);

    private static final String SELECT_COUNT = "SELECT COUNT(*) FROM `t_db_to_es` WHERE `table_index` = ?";
    private static final String ADD_INCREMENT = "INSERT INTO `t_db_to_es`(`table_index`, `increment_value`) VALUES(?, ?)";
    private static final String UPDATE_INCREMENT = "UPDATE `t_db_to_es` SET `increment_value` = ? WHERE `table_index` = ?";
    private static final String GET_INCREMENT = "SELECT `increment_value` FROM `t_db_to_es` WHERE `table_index` = ?";


    @Value("${db2es.enable:true}")
    private boolean enable;

    /** 补偿秒数 */
    @Value("${db2es.compensateSecond:60}")
    private int compensateSecond;

    private final JdbcTemplate jdbcTemplate;
    private final EsRepository esRepository;


    private String getLastValue(IncrementStorageType incrementType, String table, String incrementColumn, String index) {
        String tableColumn = getTableColumn(table, incrementColumn);
        if (ObjectUtil.isBlank(incrementType) || incrementType == IncrementStorageType.TEMP_FILE) {
            return FileUtil.read(tableColumn, index);
        } else if (incrementType == IncrementStorageType.MYSQL) {
            String name = FileUtil.fileNameOrTableKey(tableColumn, index);
            return CollectionUtil.first(jdbcTemplate.queryForList(GET_INCREMENT, String.class, name));
        } else {
            return null;
        }
    }
    private String getTableColumn(String table, String incrementColumn) {
        return table + "-" + incrementColumn;
    }
    private void saveLastValue(IncrementStorageType incrementType, String table, String incrementColumn, String index, String value) {
        String tableColumn = getTableColumn(table, incrementColumn);
        if (ObjectUtil.isBlank(incrementType) || incrementType == IncrementStorageType.TEMP_FILE) {
            FileUtil.write(tableColumn, index, value);
        } else if (incrementType == IncrementStorageType.MYSQL) {
            String name = FileUtil.fileNameOrTableKey(tableColumn, index);
            Integer count = CollectionUtil.first(jdbcTemplate.queryForList(SELECT_COUNT, Integer.class, name));
            if (ObjectUtil.greater0(count)) {
                jdbcTemplate.update(UPDATE_INCREMENT, value, name);
            } else {
                jdbcTemplate.update(ADD_INCREMENT, name, value);
            }
        }
    }

    @Async
    public Future<Long> compensateAsyncData(IncrementStorageType incrementType, Relation relation) {
        long count = 0;
        String table = relation.getTable();
        String index = relation.useIndex();
        if (ObjectUtil.isNotBlank(table) && ObjectUtil.isNotBlank(index)) {
            List<String> matchTables;
            if (relation.checkMatch()) {
                long start = System.currentTimeMillis();
                String sql = relation.matchSql();
                matchTables = jdbcTemplate.queryForList(sql, String.class);
                long sqlTime = (System.currentTimeMillis() - start);
                if (log.isDebugEnabled()) {
                    log.debug("compensate sql({}) time({}ms) return({}), size({})",
                            getSql(sql), sqlTime, CollectionUtil.toStr(matchTables), matchTables.size());
                }
            } else {
                matchTables = Collections.singletonList(table);
            }

            AtomicLong increment = new AtomicLong();
            for (String matchTable : matchTables) {
                saveSingleTable(incrementType, relation, index, matchTable, increment, true);
            }
            count = increment.get();
        }
        return new AsyncResult<>(count);
    }

    @Async
    public Future<Long> asyncData(IncrementStorageType incrementType, Relation relation) {
        String table = relation.getTable();
        String index = relation.useIndex();
        long count = 0;
        if (ObjectUtil.isNotBlank(table) && ObjectUtil.isNotBlank(index)) {
            List<String> matchTables;
            if (relation.checkMatch()) {
                long start = System.currentTimeMillis();
                String sql = relation.matchSql();
                matchTables = jdbcTemplate.queryForList(sql, String.class);
                long sqlTime = (System.currentTimeMillis() - start);
                if (log.isDebugEnabled()) {
                    log.debug("sql({}) time({}ms) return({}), size({})",
                            getSql(sql), sqlTime, CollectionUtil.toStr(matchTables), matchTables.size());
                }
            } else {
                matchTables = Collections.singletonList(table);
            }

            AtomicLong increment = new AtomicLong();
            for (String matchTable : matchTables) {
                saveSingleTable(incrementType, relation, index, matchTable, increment, false);
            }
            count = increment.get();
        }
        return new AsyncResult<>(count);
    }

    private void saveSingleTable(IncrementStorageType incrementType, Relation relation, String index,
                                 String matchTable, AtomicLong increment, boolean hasCompensate) {
        String lastValue = getLastValue(incrementType, matchTable, relation.getIncrementColumn(), index);
        boolean doCompensate = hasCompensate && (compensateSecond > 0 && ObjectUtil.isNotBlank(lastValue));
        if (doCompensate) {
            lastValue = lastValue.split(EQUALS_I_SPLIT)[0];
            Date date = DateUtil.parse(lastValue);
            if (ObjectUtil.isNotBlank(date)) {
                lastValue = DateUtil.format(DateUtil.addSecond(date, -compensateSecond), DateUtil.DateType.YYYY_MM_DD_HH_MM_SS);
            }
        }
        String matchInId = relation.matchInfo(matchTable);
        for (;;) {
            lastValue = handleGreaterAndEquals(incrementType, relation, matchTable, lastValue, matchInId, increment, doCompensate);
            if (ObjectUtil.isBlank(lastValue)) {
                return;
            }
        }
    }

    private String handleGreaterAndEquals(IncrementStorageType incrementType, Relation relation, String matchTable,
                                          String lastValue, String matchInId, AtomicLong increment, boolean hasCompensate) {
        if (!enable) {
            return null;
        }

        if (ObjectUtil.isNotBlank(lastValue) && lastValue.endsWith(EQUALS_SUFFIX)) {
            handleEquals(incrementType, relation, matchTable, lastValue, matchInId, 0, increment, false);
            return lastValue.substring(0, lastValue.length() - EQUALS_SUFFIX.length());
        }

        String sql = relation.querySql(matchTable, lastValue);
        long start = System.currentTimeMillis();
        List<Map<String, Object>> dataList = jdbcTemplate.queryForList(sql);
        if (CollectionUtil.isEmpty(dataList)) {
            return null;
        }
        long sqlTime = (System.currentTimeMillis() - start);
        if (log.isDebugEnabled()) {
            log.debug("{}sql({}) time({}ms) return size({})",
                    (hasCompensate ? "compensate " : ""), getSql(sql), sqlTime, dataList.size());
        }

        Map<String, List<Map<String, Object>>> relationData = childData(relation.getRelationMapping(), dataList);
        Map<String, List<Map<String, Object>>> nestedData = childData(relation.getNestedMapping(), dataList);
        long allSqlTime = (System.currentTimeMillis() - start);

        long esStart = System.currentTimeMillis();
        String index = relation.useIndex();
        int size = esRepository.saveDataToEs(index, fixDocument(relation, dataList, matchInId, relationData, nestedData));
        increment.addAndGet(size);
        long end = System.currentTimeMillis();
        if (log.isInfoEnabled()) {
            log.info("{}greater({}) sql time({}ms) size({}) batch to({}) time({}ms) success({}), all time({}ms)",
                    (hasCompensate ? "compensate " : ""), lastValue, allSqlTime, dataList.size(), index, (end - esStart), size, (end - start));
        }
        if (size == 0) {
            return null;
        }

        lastValue = getLast(relation, dataList);
        if (ObjectUtil.isBlank(lastValue)) {
            return null;
        }
        handleEquals(incrementType, relation, matchTable, lastValue, matchInId, 0, increment, hasCompensate);
        saveLastValue(incrementType, matchTable,
                relation.getIncrementColumn() + (hasCompensate ? "=compensate" : ""), index, lastValue);

        if (dataList.size() < relation.getLimit()) {
            return null;
        }
        return lastValue;
    }
    private void handleEquals(IncrementStorageType incrementType, Relation relation, String matchTable,
                              String tempColumnValue, String matchInId, int lastEqualsCount,
                              AtomicLong increment, boolean hasCompensate) {
        if (!enable) {
            return;
        }

        String eai;
        if (tempColumnValue.endsWith(EQUALS_SUFFIX)) {
            eai = tempColumnValue.substring(0, tempColumnValue.length() - EQUALS_SUFFIX.length());
        } else {
            eai = tempColumnValue;
        }
        String[] equalsValueArr = eai.split(EQUALS_I_SPLIT);
        String equalsValue = equalsValueArr[0];
        long nowMs = System.currentTimeMillis();

        // 上面: time > '2010-10-10 00:00:01' | 1286640001000, 当前: time = '2010-10-10 00:00:01' | 1286640001000
        String equalsCountSql = relation.equalsCountSql(matchTable, equalsValue);
        long start = System.currentTimeMillis();
        Integer equalsCount = CollectionUtil.first(jdbcTemplate.queryForList(equalsCountSql, Integer.class));
        if (log.isDebugEnabled()) {
            log.debug("{}equals count sql({}) time({}ms) return({})",
                    (hasCompensate ? "compensate " : ""), getSql(equalsCountSql), (System.currentTimeMillis() - start), equalsCount);
        }
        if (ObjectUtil.less0(equalsCount)) {
            currentSecondHandle(equalsValue, nowMs, incrementType, relation,
                    matchTable, 0, matchInId, 0, increment, hasCompensate);
            return;
        }
        if (lastEqualsCount > 0 && lastEqualsCount == equalsCount) {
            currentSecondHandle(equalsValue, nowMs, incrementType, relation,
                    matchTable, 0, matchInId, equalsCount, increment, hasCompensate);
            return;
        }

        String index = relation.useIndex();
        int equalsLoopCount = relation.loopCount(equalsCount);
        int i = 0;
        if (equalsValueArr.length == 2) {
            i = ObjectUtil.toInt(equalsValueArr[1]);
            // if count = 1000, limit = 10, save has 101
            if (i * relation.getLimit() > equalsCount) {
                currentSecondHandle(equalsValue, nowMs, incrementType, relation,
                        matchTable, 0, matchInId, equalsCount, increment, hasCompensate);
                return;
            }
            if (i < 0) {
                i = 0;
            }
        }
        for (; i < equalsLoopCount; i++) {
            String equalsSql = relation.equalsQuerySql(matchTable, equalsValue, i);
            long sqlStart = System.currentTimeMillis();
            List<Map<String, Object>> equalsDataList = jdbcTemplate.queryForList(equalsSql);
            if (CollectionUtil.isEmpty(equalsDataList)) {
                currentSecondHandle(equalsValue, nowMs, incrementType, relation,
                        matchTable, i, matchInId, equalsCount, increment, hasCompensate);
                return;
            }
            long sqlTime = (System.currentTimeMillis() - sqlStart);
            if (log.isDebugEnabled()) {
                log.debug("{}equals sql({}) time({}ms) return size({})",
                        (hasCompensate ? "compensate " : ""), getSql(equalsSql), sqlTime, equalsDataList.size());
            }

            Map<String, List<Map<String, Object>>> relationData = childData(relation.getRelationMapping(), equalsDataList);
            Map<String, List<Map<String, Object>>> nestedData = childData(relation.getNestedMapping(), equalsDataList);
            long allSqlTime = (System.currentTimeMillis() - sqlStart);

            long esStart = System.currentTimeMillis();
            int size = esRepository.saveDataToEs(index, fixDocument(relation, equalsDataList, matchInId, relationData, nestedData));
            increment.addAndGet(size);
            long end = System.currentTimeMillis();
            if (log.isInfoEnabled()) {
                log.info("{}equals({}-{}: {}) sql time({}ms) size({}) batch to({}) time({}ms) success({}), all time({}ms)",
                        (hasCompensate ? "compensate " : ""), equalsValue, i, equalsLoopCount, allSqlTime,
                        equalsDataList.size(), index, (end - esStart), size, (end - sqlStart));
            }

            if ((size == 0) || (equalsDataList.size() < relation.getLimit())) {
                // 如果时间戳是在当前秒, 则休眠到下一秒, 并再执行 = 操作, 避免出现同步时当前秒还没过去却不再处理, 但是数据库却可能在当前秒后面的毫秒中操作数据
                currentSecondHandle(equalsValue, nowMs, incrementType, relation,
                        matchTable, i, matchInId, equalsCount, increment, hasCompensate);
                return;
            } else {
                String valueToSave = equalsValue + EQUALS_I_SPLIT + i + EQUALS_SUFFIX;
                saveLastValue(incrementType, matchTable,
                        relation.getIncrementColumn() + (hasCompensate ? "=compensate" : ""), index, valueToSave);
            }
        }
    }

    /**
     * 如果增量的数据是当前秒, 则休眠到下一秒并再次执行上面的 = 操作,
     * 这样做是为了避免同步时在当前秒已经没有处理要写(比如当前秒前面的毫秒时)然后就跳过了,
     * 但到了当前秒的后面的毫秒时, 数据库(尤其 mysql 在处理毫秒时还会自动进行四舍五入)又操作了数据.
     *
     * 如果不这样处理, 当前秒后面的毫秒数的时候数据库操作的记录会无法同步
     */
    private void currentSecondHandle(String equalsValue, long nowMs, IncrementStorageType incrementType,
                                     Relation relation, String matchTable, int i, String matchInId,
                                     int currentEqualsCount, AtomicLong increment, boolean hasCompensate) {
        Date equalsDate = DateUtil.parse(equalsValue);
        if (ObjectUtil.isNotBlank(equalsDate)) {
            long equalsMs = equalsDate.getTime();
            boolean needSleepToNextSecond = (nowMs > equalsMs) && (equalsMs / 1000 == nowMs / 1000);
            if (needSleepToNextSecond) {
                try {
                    long nextSecondMs = ((nowMs / 1000) + 1) * 1000;
                    Thread.sleep(nextSecondMs - nowMs);
                    handleEquals(incrementType, relation, matchTable, equalsValue + EQUALS_I_SPLIT + i,
                            matchInId, currentEqualsCount, increment, hasCompensate);
                } catch (InterruptedException e) {
                    if (log.isErrorEnabled()) {
                        log.error("增量值是当前秒, 休眠到下一秒时异常", e);
                    }
                }
            }
        }
    }

    private Map<String, List<Map<String, Object>>> childData(Map<String, ChildMapping> childMapping, List<Map<String, Object>> dataList) {
        if (CollectionUtil.isEmpty(dataList)) {
            return Collections.emptyMap();
        }

        Map<String, List<Map<String, Object>>> returnMap = Maps.newHashMap();
        if (CollectionUtil.isNotEmpty(childMapping)) {
            for (Map.Entry<String, ChildMapping> entry : childMapping.entrySet()) {
                String key = entry.getKey();
                ChildMapping nested = entry.getValue();

                List<Object> relations = Lists.newArrayList();
                for (Map<String, Object> data : dataList) {
                    relations.add(data.get(nested.getMainField()));
                }
                if (CollectionUtil.isNotEmpty(relations)) {
                    String sql = nested.nestedQuerySql(relations);
                    if (ObjectUtil.isNotBlank(sql)) {
                        long start = System.currentTimeMillis();
                        List<Map<String, Object>> nestedDataList = jdbcTemplate.queryForList(sql);
                        if (log.isDebugEnabled()) {
                            log.debug("child({}) sql({}) time({}ms) return size({})",
                                    key, getSql(sql), (System.currentTimeMillis() - start), nestedDataList.size());
                        }
                        returnMap.put(key, nestedDataList);
                    }
                }
            }
        }
        return returnMap;
    }

    private String getSql(String sql) {
        if (ObjectUtil.isBlank(sql)) {
            return ObjectUtil.EMPTY;
        }
        int max = 200;
        if (sql.length() <= max) {
            return sql;
        }
        String fromSql = sql.substring(sql.toUpperCase().indexOf(" FROM ") + 1);
        if (fromSql.length() <= max) {
            return fromSql;
        }
        String[] valueArr = fromSql.split(",");
        int len = valueArr.length;
        if (len <= 10) {
            return fromSql;
        }
        return Joiner.on(", ").join(Arrays.asList(
                valueArr[0], valueArr[1], valueArr[2],
                " ... ",
                valueArr[len - 3], valueArr[len - 2], valueArr[len - 1]
        ));
    }
    private String getLast(Relation relation, List<Map<String, Object>> dataList) {
        Map<String, Object> last = CollectionUtil.last(dataList);
        if (CollectionUtil.isNotEmpty(last)) {
            String column = relation.getIncrementColumn();
            Object obj = last.get(column.contains(".") ? column.substring(column.indexOf(".") + 1) : column);
            if (ObjectUtil.isNotBlank(obj)) {
                String lastData;
                if (obj instanceof Date) {
                    lastData = DateUtil.format((Date) obj, DateUtil.DateType.YYYY_MM_DD_HH_MM_SS);
                } else {
                    lastData = obj.toString();
                }
                return lastData;
            }
        }
        return null;
    }
    private Map<String, Map<String, String>> fixDocument(Relation relation, List<Map<String, Object>> dataList, String matchInId,
                                                         Map<String, List<Map<String, Object>>> relationData,
                                                         Map<String, List<Map<String, Object>>> nestedData) {
        Map<String, Map<String, Map<String, Object>>> relationMap = Maps.newHashMap();
        if (CollectionUtil.isNotEmpty(relation.getRelationMapping())) {
            for (Map.Entry<String, ChildMapping> entry : relation.getRelationMapping().entrySet()) {
                String key = entry.getKey();
                ChildMapping child = entry.getValue();

                List<Map<String, Object>> list = relationData.get(key);
                if (CollectionUtil.isNotEmpty(list)) {
                    String tableField = child.getChildField();
                    Map<String, Map<String, Object>> dataMap = Maps.newHashMap();
                    for (Map<String, Object> data : list) {
                        String fieldData = ObjectUtil.toStr(data.get(tableField));
                        if (ObjectUtil.isNotBlank(fieldData)) {
                            data.remove(tableField);
                            dataMap.put(fieldData, data);
                        }
                    }
                    relationMap.put(key, dataMap);
                }
            }
        }
        Map<String, Multimap<String, Map<String, Object>>> nestedMap = Maps.newHashMap();
        if (CollectionUtil.isNotEmpty(relation.getNestedMapping())) {
            for (Map.Entry<String, ChildMapping> entry : relation.getNestedMapping().entrySet()) {
                String key = entry.getKey();
                ChildMapping nested = entry.getValue();

                List<Map<String, Object>> list = nestedData.get(key);
                if (CollectionUtil.isNotEmpty(list)) {
                    String tableField = nested.getChildField();
                    Multimap<String, Map<String, Object>> multiMap = LinkedHashMultimap.create();
                    for (Map<String, Object> data : list) {
                        String fieldData = ObjectUtil.toStr(data.get(tableField));
                        if (ObjectUtil.isNotBlank(fieldData)) {
                            data.remove(tableField);
                            multiMap.put(fieldData, data);
                        }
                    }
                    nestedMap.put(key, multiMap);
                }
            }
        }

        Map<String, Map<String, String>> documents = Maps.newHashMap();
        for (Map<String, Object> data : dataList) {
            StringBuilder idBuild = new StringBuilder();
            String idPrefix = relation.getIdPrefix();
            if (ObjectUtil.isNotBlank(idPrefix)) {
                idBuild.append(idPrefix.trim());
            }
            if (relation.isPatternToId() && ObjectUtil.isNotBlank(matchInId)) {
                if (idBuild.length() > 0) {
                    idBuild.append("-");
                }
                idBuild.append(matchInId.trim());
            }
            for (String column : relation.getKeyColumn()) {
                if (idBuild.length() > 0) {
                    idBuild.append("-");
                }
                String str = ObjectUtil.toStr(data.get(column));
                if (ObjectUtil.isNotBlank(str)) {
                    idBuild.append(str.trim());
                }
            }
            String idSuffix = relation.getIdSuffix();
            if (ObjectUtil.isNotBlank(idSuffix)) {
                if (idBuild.length() > 0) {
                    idBuild.append("-");
                }
                idBuild.append(idSuffix.trim());
            }
            String id = idBuild.length() == 0 ? UUIDs.base64UUID() : idBuild.toString();

            if (CollectionUtil.isNotEmpty(relation.getRelationMapping())) {
                for (Map.Entry<String, ChildMapping> entry : relation.getRelationMapping().entrySet()) {
                    String nestedKey = entry.getKey();
                    if (data.containsKey(nestedKey)) {
                        if (log.isWarnEnabled()) {
                            log.warn("one to one mapping({}) has already alias in primary sql, ignore put)", nestedKey);
                        }
                    } else {
                        ChildMapping nestedValue = entry.getValue();
                        Map<String, Map<String, Object>> dataMap = relationMap.get(nestedKey);
                        if (CollectionUtil.isNotEmpty(dataMap)) {
                            Map<String, Object> map = dataMap.get(ObjectUtil.toStr(data.get(nestedValue.getMainField())));
                            if (CollectionUtil.isNotEmpty(map)) {
                                for (Map.Entry<String, Object> me : map.entrySet()) {
                                    if (data.containsKey(me.getKey())) {
                                        if (log.isWarnEnabled()) {
                                            log.warn("one to one mapping({}) field({}) has already alias in primary sql, ignore put)",
                                                    nestedKey, me.getKey());
                                        }
                                    } else {
                                        data.put(me.getKey(), me.getValue());
                                    }
                                }
                            }
                        }
                    }
                }
            }

            if (CollectionUtil.isNotEmpty(relation.getNestedMapping())) {
                for (Map.Entry<String, ChildMapping> entry : relation.getNestedMapping().entrySet()) {
                    String nestedKey = entry.getKey();
                    if (data.containsKey(nestedKey)) {
                        if (log.isWarnEnabled()) {
                            log.warn("nested mapping({}) has already alias in primary sql, ignore put)", nestedKey);
                        }
                    } else {
                        ChildMapping nestedValue = entry.getValue();
                        Multimap<String, Map<String, Object>> multimap = nestedMap.get(nestedKey);
                        if (ObjectUtil.isNotBlank(multimap) && multimap.size() > 0) {
                            Collection<Map<String, Object>> list = multimap.get(ObjectUtil.toStr(data.get(nestedValue.getMainField())));
                            if (CollectionUtil.isNotEmpty(list)) {
                                data.put(nestedKey, list);
                            }
                        }
                    }
                }
            }

            Map<String, Object> dataMap = Maps.newHashMap();
            for (Map.Entry<String, Object> entry : data.entrySet()) {
                String key = relation.useField(entry.getKey());
                if (ObjectUtil.isNotBlank(key)) {
                    Object value = entry.getValue();
                    if (ObjectUtil.isNotBlank(value)) {
                        if (Sets.newHashSet("0000-00-00", "00:00:00", "0000-00-00 00:00:00").contains(value.toString())) {
                            dataMap.put(key, NIL_DATE_TIME);
                        } else {
                            dataMap.put(key, value);
                        }
                    } else {
                        dataMap.put(key, "");
                    }
                }
            }

            if (CollectionUtil.isNotEmpty(dataMap)) {
                Map<String, String> sourceMap = Maps.newHashMap();
                sourceMap.put("data", JsonUtil.toJson(dataMap));

                if (CollectionUtil.isNotEmpty(relation.getRouteColumn())) {
                    List<String> routes = Lists.newArrayList();
                    for (String route : relation.getRouteColumn()) {
                        Object obj = data.get(route);
                        if (ObjectUtil.isNotBlank(obj)) {
                            routes.add(ObjectUtil.toStr(obj).trim());
                        }
                    }
                    if (CollectionUtil.isNotEmpty(routes)) {
                        sourceMap.put("routing", CollectionUtil.toStr(routes));
                    }
                }
                if (ObjectUtil.isNotBlank(relation.getVersionColumn())) {
                    String version = ObjectUtil.toStr(data.get(relation.getVersionColumn()));
                    if (ObjectUtil.isNumber(version) && ObjectUtil.greater0(Double.parseDouble(version))) {
                        sourceMap.put("version", version);
                    } else {
                        Date datetime = DateUtil.parse(version);
                        if (ObjectUtil.isNotBlank(datetime)) {
                            sourceMap.put("version", ObjectUtil.toStr(datetime.getTime()));
                        }
                    }
                }
                documents.put(id, sourceMap);
            }
        }
        if (documents.size() < dataList.size()) {
            if (log.isWarnEnabled()) {
                log.warn("data size({}) <--> es size({}), may be has duplicate id", dataList.size(), documents.size());
            }
        }
        return documents;
    }
}
